package com.heima.kafaka.sample;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;

/**
 * <p>
 * 生产着
 * </p>
 *
 * @author lzt
 * @version 1.0
 * @since 2024/4/15 下午10:03
 */
public class ConsumerQuickStart {
    public static void main(String[] args) {
        // 1. kafka配置信息
        Properties prop = new Properties();
        //kafka连接地址
        prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        //key和value的序列化
        prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");


        // 2.创建kafka生产对象
        KafkaProducer<String, String> producer = new KafkaProducer<String,String>(prop);

        // 3.发送信息

        ProducerRecord<String, String> kvProduceer = new ProducerRecord<String, String>("topic-first","key-001","hello-kafka");
        // 同步发送信息
//        producer.send(kvProduceer);

        // 异步信息发送
        producer.send(kvProduceer, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if(e != null) {
                    System.out.println("出现异常");
                }
                System.out.println(recordMetadata.offset());
            }
        });

        // 4.关闭信息通过 不然发送不成功
        producer.close();
    }
}
